package main.scala.demo

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * SparkStreamContextCheckPoinDemo
  *  updateStateByKey API 方法实现
  * @author zhangyimin
  *         2018-10-17 下午12:13
  * @version 1.0
  */
object SparkStreamContextCheckPointDemo {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
    val ssc = StreamingContext.getOrCreate("hdfs://10.16.7.36:9000/DStream03", createStreamingContext)
    ssc.start()
    ssc.awaitTermination()
  }


  def createStreamingContext(): StreamingContext = {
    val sparkConf = new SparkConf().setAppName("MyNetworkWordCount").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.checkpoint("hdfs://10.16.7.36:9000/DStream03")
    val updateFunc = (values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.sum

      val previousCount = state.getOrElse(0)

      Some(currentCount + previousCount)
    }
    // Create a ReceiverInputDStream on target ip:port and count the
    // words in input stream of \n delimited test (eg. generated by 'nc')
    val lines = ssc.socketTextStream("10.16.7.36", 5678)
    val words = lines.flatMap(_.split(" "))
    val wordDStream = words.map(x => (x, 1))

    // Update the cumulative count using updateStateByKey
    // This will give a Dstream made of state (which is the cumulative count of the words)
    val stateDStream = wordDStream.updateStateByKey(updateFunc)
    stateDStream.print()
    ssc
  }


}

